package oktx.config;

import oktx.participant.Participant;
import oktx.RemoteType;
import oktx.terminator.Terminator;
import oktx.TransactionContext;
import oktx.thread.TransactionThreadDetach;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;

import javax.transaction.SystemException;
import java.util.List;

/**
 * Created by huangdachao on 2018/5/16 21:01.
 * 本地事务提交或回滚，自动异步提交或回滚远程调用的子事务
 */
public class LocalTxTerminator {
    private static final Logger LOG = LoggerFactory.getLogger(LocalTxTerminator.class);
    @Autowired TaskExecutor taskExecutor;
    @Autowired List<Terminator> terminators;
    @Autowired(required = false) TransactionThreadDetach threadDetach;

    /**
     *
     * @param xid
     * @param local 是否本地请求回滚事务
     */
    public void rollback(String xid, boolean local) {
        TransactionContext tc = TransactionContext.getContext(xid);
        if (tc != null) {
            if (!local && tc.isRoot()) {
                return; // 远程参与者回头回滚事务发起者，直接忽略
            }
            try {
                tc.getTransaction().rollback();
                TransactionContext.clearContext(xid);
                if (threadDetach != null) {
                    threadDetach.detach(tc.getTransaction());
                }

                tc.getParticipants().forEach(p -> {
                    if (p.getType() == RemoteType.consumer) {
                        return; // 忽略调用方
                    }
                    taskExecutor.execute(() -> selectTerminator(p).rollback(p));
                });
                LOG.debug("JTA事务回滚：" + tc);
            } catch (SystemException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     *
     * @param xid
     * @param local 是否本地请求提交事务
     */
    public void commit(String xid, boolean local) {
        TransactionContext tc = TransactionContext.getContext(xid);
        if (tc != null) {
            if (!local && tc.isRoot()) {
                LOG.debug("JTA事务提交取消");
                return; // 远程参与者回头提交事务发起者，直接忽略
            }
            try {
                tc.getTransaction().commit();
                TransactionContext.clearContext(xid);
                if (threadDetach != null) {
                    threadDetach.detach(tc.getTransaction());
                }

                tc.getParticipants().forEach(p -> {
                    if (p.getType() == RemoteType.consumer) {
                        return; // 忽略调用方
                    }
                    taskExecutor.execute(() -> selectTerminator(p).commit(p));
                });
                LOG.debug("JTA事务提交：" + tc);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }

    @SuppressWarnings("unchecked")
    private Terminator selectTerminator(Participant p) {
        Terminator t = terminators.stream().filter(term -> term.support(p)).findFirst().orElse(null);
        if (t == null) {
            throw new RuntimeException("没有找到支持" + p.getClass().getName() + "的Terminator");
        }
        return t;
    }
}
